package com.atguigu.sink;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Created by RUI on 2021/6/10 19:54
 */
public class MySink {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream("hadoop162",9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],Long.valueOf(split[1]),Integer.valueOf(split[2]));
                    }
                })
                .addSink(new MySinkFunction());

        env.execute();
    }

    private static class MySinkFunction extends RichSinkFunction<WaterSensor>{

        PreparedStatement ps;
        Connection connection;

        @Override
        public void open(Configuration parameters) throws Exception {

            connection = DriverManager.getConnection("jdbc:mysql://hadoop162:3306/test", "root", "aaaaaa");

            String sql = "replace into sensor values(?,?,?)";
            ps = connection.prepareStatement(sql);
        }

        @Override
        public void invoke(WaterSensor ws, Context context) throws Exception {

            ps.setString(1,ws.getId());
            ps.setLong(2,ws.getTs());
            ps.setInt(3,ws.getVc());

            ps.execute();

        }

        @Override
        public void close() throws Exception {
            if (null != ps) {
                ps.close();
            }

            if (null != connection) {
                connection.close();
            }
        }
    }
}
